package com.gitee.Transformation

import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
import org.apache.flink.api.scala.ExecutionEnvironment

/*
  这个很重要, 类似于select的join
  join操作
  左外连接
  右外连接
  全外连接
 */
object JoinTest {
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._
    env.setParallelism(8)
    case class Sorce(id: String, name: String, subjectId: String, sorce: String)
    val sorceTable: DataSet[Sorce] = env.readCsvFile[Sorce]("file:///D:\\data\\input\\score.csv")
    case class Subject(subjectId: String, subjectName: String)
    val subjectTable: DataSet[Subject] = env.readCsvFile[Subject]("file:///D:\\data\\input\\subject.csv")

    // join
    // 根据sorceTable中索引为2, subjectTable中索引为0的两个表关联起来
    val result01: DataSet[(String, String, String, String)] = sorceTable.join(subjectTable).where(2).equalTo(0).apply((left, right) => {
      (left.id, left.name, right.subjectName, left.sorce)
    })
    result01.printOnTaskManager("01>>>")

    // 左外连接
    //JoinHint.OPTIMIZER_CHOOSES默认值,表示由Flink优化器选择join的方式
    //BROADCAST_HASH_FIRST：广播第一个输入端，同时基于它构建一个哈希表，而第二个输入端作为探索端，选择这种策略的场景是第一个输入端规模很小；
    //BROADCAST_HASH_SECOND：广播第二个输入端并基于它构建哈希表，第一个输入端作为探索端，选择这种策略的场景是第二个输入端的规模很小；
    //REPARTITION_HASH_FIRST：该策略会导致两个输入端都会被重分区，但会基于第一个输入端构建哈希表。该策略适用于第一个输入端数据量小于第二个输入端的数据量，但这两个输入端的规模仍然很大，优化器也是当没有办法估算大小，没有已存在的分区以及排序顺序可被使用时系统默认采用的策略；
    //REPARTITION_HASH_SECOND：该策略会导致两个输入端都会被重分区，但会基于第二个输入端构建哈希表。该策略适用于两个输入端的规模都很大，但第二个输入端的数据量小于第一个输入端的情况；
    //REPARTITION_SORT_MERGE：输入端被以流的形式进行连接并合并成排过序的输入。该策略适用于一个或两个输入端都已排过序的情况；
    // 对于这些选择器我们不需要去实现,Flink自己会选择最优的,但是如果我们发现Flink和我们想要的选择符合,可以自己手动选择
    val joinFunctionAssigner01: JoinFunctionAssigner[Sorce, Subject] = sorceTable.leftOuterJoin(subjectTable,JoinHint.OPTIMIZER_CHOOSES).where(2).equalTo(0)
    val result02: DataSet[(String, String, String, String)] = joinFunctionAssigner01.apply((left, right) => {
      if (right == null) {
        (left.id, left.name, "null", left.sorce)
      } else {
        (left.id, left.name, right.subjectName, left.sorce)
      }
    })
    result02.printOnTaskManager("02>>>")

    // 右外连接
    val joinFunctionAssigner02: JoinFunctionAssigner[Sorce, Subject] = sorceTable.rightOuterJoin(subjectTable, JoinHint.BROADCAST_HASH_FIRST).where(2).equalTo(0)
    val result03: DataSet[(String, String, String)] = joinFunctionAssigner02.apply((left, right) => {
      if (left == null) {
        ("null", right.subjectName, "0")
      } else {
        (left.name, right.subjectName, left.sorce)
      }
    })
    result03.printOnTaskManager("03>>>")

    // 全外连接
    val joinFunctionAssigner03: JoinFunctionAssigner[Sorce, Subject] = sorceTable.fullOuterJoin(subjectTable).where(2).equalTo(0)
    joinFunctionAssigner03.apply((x,y)=>{
      if (x==null){

      }else if(y==null){

      }else{
        (x.name,x.sorce,y.subjectName)
      }
    }).printOnTaskManager("04>>>")
  }

}
